
package com.ccnu.demojava.net.nio;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * @author raohj
 * @date 2018年10月30日
 * @version 
 */
public class RNioServer {
	private volatile static boolean stop = false;

	/**
	 * @author raohj
	 * @date 2018年10月30日
	 * @version 
	 */
	public class RTimeTask implements Runnable{
        public SelectionKey sk;
        public RTimeTask(SelectionKey key) {
            this.sk = key;
        }

        /**
         * When an object implementing interface <code>Runnable</code> is used
         * to create a thread, starting the thread causes the object's
         * <code>run</code> method to be called in that separately executing
         * thread.
         * <p>
         * The general contract of the method <code>run</code> is that it may
         * take any action whatsoever.
         *
         * @see Thread#run()
         */
        public void run() {
            try{
                if(sk.isValid()){
                	// 有客户端连接进来
                    if(sk.isAcceptable()){
                        SocketChannel clnChan = ((ServerSocketChannel)sk.channel()).accept();
                        clnChan.configureBlocking(false);
                        clnChan.socket().setReuseAddress(false);
                        clnChan.register(sk.selector(), SelectionKey.OP_READ);
                    }
                    // 有数据可读
                    if(sk.isReadable()){
                        SocketChannel sc = (SocketChannel) sk.channel();
                        ByteBuffer readBuffer = ByteBuffer.allocate(1024);
                        int readByte = sc.read(readBuffer);
                        if(readByte>0){
                            readBuffer.clear();
                            readBuffer.put(new String("[" + Thread.currentThread().threadId() + "]" + "this is my Server, Please deal later\n").getBytes());
                            readBuffer.flip();
                            sc.write(readBuffer);
                            sk.channel();
                            sc.close();
                        }
                        else if( readByte<0) {
                            sk.channel();
                            sc.close();
                        }
                        else {

                        }
                    }
                }
            }catch (Exception e){

            }
        }
    }
    // 异步单线程
    public void testNioServer1(){
        try {
            // 选择器
            Selector seltor = Selector.open();
            // 通道
            ServerSocketChannel server = ServerSocketChannel.open();
            // 监听地址
            InetSocketAddress isa  = new InetSocketAddress(30001);
            server.socket().bind(isa, 1024);	// 绑定监听的地址(完成绑定+监听)
            server.configureBlocking(false);	// 通道设置为非堵塞模式
            server.register(seltor, SelectionKey.OP_ACCEPT);	// 注册到Reactor线程的多路复用器上，并监听ACCEPT事件
            System.out.println("listen on port 30001");
            while(true){
                // 设置select的超时时间
                int n =  seltor.select();
                if(n==0) continue;
                Iterator<?> ite = seltor.selectedKeys().iterator();
                while(ite.hasNext()) {
                    SelectionKey key = (SelectionKey) ite.next();
                    // 获取到某个事件，并将其移除
                    ite.remove();
                    try {
                        handInput(key);
                    } catch (Exception e) {
                        if (key != null) {
                            key.cancel();
                            if (key.channel() != null) {
                                key.channel().close();
                            }
                        }
                    }
                }
            }
        }
        catch (IOException e){
            e.printStackTrace();
        }
        return;
    }

    // 异步多线程
    public void testNioServer2(){
        try {
            // 选择器
            Selector seltor = Selector.open();
            // 通道
            ServerSocketChannel server = ServerSocketChannel.open();
            // 监听地址
            InetSocketAddress isa  = new InetSocketAddress(30001);
            server.socket().bind(isa, 1024);
            server.configureBlocking(false);
            server.register(seltor, SelectionKey.OP_ACCEPT);
            System.out.println("listen on port 30003");
            while(!RNioServer.getStop()){
                // 设置select的超时时间
                int n =  seltor.select();
                if(n==0) continue;
                Iterator<?> ite = seltor.selectedKeys().iterator();
                while(ite.hasNext() && !RNioServer.getStop()) {
                    SelectionKey key = (SelectionKey) ite.next();
                    // 获取到某个事件，并将其移除
                    ite.remove();
                    try {
                        new Thread(new RTimeTask(key)).start();
                    } catch (Exception e) {
                        if (key != null) {
                            key.cancel();
                            if (key.channel() != null) {
                                key.channel().close();
                            }
                        }
                    }
                }
            }
        }
        catch (IOException e){
            e.printStackTrace();
        }
        return;
    }

    // 异步线程池
    public void testNioServer3(){
        try {
            ExecutorService testPool = Executors.newFixedThreadPool(100);
            // 选择器
            Selector seltor = Selector.open();
            // 通道
            ServerSocketChannel server = ServerSocketChannel.open();
            // 监听地址
            InetSocketAddress isa  = new InetSocketAddress(30001);
            server.socket().bind(isa, 1024);
            server.configureBlocking(false);
            server.register(seltor, SelectionKey.OP_ACCEPT);
            System.out.println("listen on port 30003");
            while(!RNioServer.getStop()){
                // 设置select的超时时间
                int n =  seltor.select();
                if(n==0) continue;
                Iterator<?> ite = seltor.selectedKeys().iterator();
                while(ite.hasNext()) {
                    SelectionKey key = (SelectionKey) ite.next();
                    // 获取到某个事件，并将其移除
                    ite.remove();
                    try {
                        testPool.submit(new RTimeTask(key));
                    } catch (Exception e) {
                        if (key != null) {
                            key.cancel();
                            if (key.channel() != null) {
                                key.channel().close();
                            }
                        }
                    }
                }
            }
            
            if(seltor!=null) {
            	try {
            		// selector 被关闭以后，其上的channel、Pipe等资源均会被关闭
            		seltor.close();
            	}catch (Exception e) {
					e.printStackTrace();
				}
            }
        }
        catch (IOException e){
            e.printStackTrace();
        }
        return;
    }

    private void handInput(SelectionKey key) throws IOException{
        if(key.isValid()){
        	// 如果是连接请求
            if(key.isAcceptable()){
            	ServerSocketChannel serChan = (ServerSocketChannel) key.channel();
                SocketChannel clnChan = serChan.accept();
                clnChan.configureBlocking(false);
                clnChan.socket().setReuseAddress(false);
                clnChan.register(key.selector(), SelectionKey.OP_READ);
//                clnChan.register(key.selector(), SelectionKey.OP_WRITE);
//                clnChan.register(key.selector(), SelectionKey.OP_CONNECT);
            }
            // 如果是读请求
            if(key.isReadable()){
                SocketChannel sc = (SocketChannel) key.channel();
                ByteBuffer readBuffer = ByteBuffer.allocate(1024);
                int readByte = sc.read(readBuffer);
                if(readByte>0){
                    readBuffer.clear();
                    readBuffer.put(new String("[" + Thread.currentThread().threadId() + "]" + "this is my Server, Please deal later\n").getBytes());
                    readBuffer.flip();
                    sc.write(readBuffer);
                    key.channel().close();
                    sc.close();
                }
                else if( readByte<0) {
                	// 对端链路关闭，取消对此通道的监听
                	key.cancel();
                    sc.close();
                }
                else { // readByte == 0
                	// 已经读取完毕了
                }
            }
        }
    }
    
    // 写入通道
    private void doWrite(SocketChannel channel, String resp) 
    	throws IOException{
    	if(resp != null && resp.trim().length()>0) {
    		byte [] bytes = resp.getBytes();
    		ByteBuffer writeBuffer = ByteBuffer.allocate(bytes.length);
    		writeBuffer.put(bytes);
    		writeBuffer.flip();
    		channel.write(writeBuffer);
    	}
    }

	public static boolean getStop() {
		return stop;
	}

	public static void setStop(boolean stop) {
		RNioServer.stop = stop;
	}
}
